package com.lujieni.nio.c4;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class MultiThreadServer {

    public static void main(String[] args) throws IOException, InterruptedException {
        // 1.初始化selector
        Selector selector = Selector.open();

        ServerSocketChannel server = ServerSocketChannel.open();
        server.bind(new InetSocketAddress(8080)); // 监听8080端口
        server.configureBlocking(false); // 必须要加否则报错

        // 2.和selector建立绑定并指定监听的事件
        SelectionKey sscKey = server.register(selector, 0, null);// 注册到selector中并且不监听任何事件
        sscKey.interestOps(SelectionKey.OP_ACCEPT);

        Worker worker = new Worker("work");
        Worker[] workers = new Worker[2];

        for (int i = 0; i < 2 ; i++) {
            workers[i] = new Worker("work"+i);
        }

        AtomicInteger at = new AtomicInteger(0);

        while (true) {
            // 3.阻塞等到消费事件
            selector.select(); // 阻塞
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()){
                SelectionKey item = iterator.next();
                iterator.remove();
                if(item.isAcceptable()){
                    SocketChannel channel = server.accept();
                    workers[at.getAndIncrement() % 2].register(channel);
                }
            }
        }

    }

}

class Worker implements Runnable{
    private Thread thread;

    private Selector selector;

    private ArrayBlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(100);

    public Worker(String threadName) throws IOException {
        selector = Selector.open();
        thread = new Thread(this, threadName);
        thread.start(); // happen-before selector的初始化一定优先于开启线程
    }

    public void register(SocketChannel channel) throws ClosedChannelException, InterruptedException {
        queue.put(()->{
            try {
                channel.configureBlocking(false);
                channel.register(selector, SelectionKey.OP_READ, null);
            } catch (ClosedChannelException e) {
                throw new RuntimeException(e);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        });
        selector.wakeup();
    }

    @Override
    public void run() {
        while (true){
            try {
                Runnable task = queue.poll();
                if(task != null){
                    task.run();
                }
                selector.select();
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()){
                    SelectionKey item = iterator.next();
                    iterator.remove();
                    if (item.isReadable()) {
                        try {
                            SocketChannel channel = (SocketChannel)item.channel();
                            ByteBuffer bb = ByteBuffer.allocate(16);
                            channel.read(bb);
                            bb.flip();
                            System.out.println(Thread.currentThread().getName()+":"+StandardCharsets.UTF_8.decode(bb));
                        } catch (IOException e) {
                            item.cancel();
                        }
                    }
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }


        }
    }
}
